package com.deepglint.tableapi.udf;

import com.deepglint.beans.SensorReading;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-28 19:28
 * 自定义聚合函数
 */
public class UDFTest_AggregateFunction {
    public static void main(String[] args) throws Exception {
        // 1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2.读取数据
        String path = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStreamSource<String> sourceStream = env.readTextFile(path);

        // 3.转换pojo
        SingleOutputStreamOperator<SensorReading> mapStream = sourceStream.map(line -> {
            String[] split = line.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        // 4.将流转换为表
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Table sensorTable = tableEnv.fromDataStream(mapStream, "id,timestamp as ts,temperature as temp");
        tableEnv.createTemporaryView("sensor", sensorTable);

        // 5. 自定义的table函数，实现id拆分，输出(word，length)
        // 5.1 table api

        // 将函数注册到环境中
        AvgTemp avgTemp = new AvgTemp();
        tableEnv.registerFunction("avgTemp", avgTemp);
        Table resultTable = sensorTable
                .groupBy("id")
                .aggregate("avgTemp(temp) as avgtemp")
                .select("id,avgtemp");

        // 5.2
        Table sqlQuery = tableEnv.sqlQuery("select id,avgTemp(temp) from sensor group by id");


        // 输出
        tableEnv.toRetractStream(resultTable, Row.class).print("table");
        tableEnv.toRetractStream(sqlQuery, Row.class).print("sql");

        env.execute();
    }

    public static class AvgTemp extends AggregateFunction<Double, Tuple2<Double, Integer>> {

        @Override
        public Double getValue(Tuple2<Double, Integer> accumulator) {
            return accumulator.f0 / accumulator.f1;
        }

        @Override
        public Tuple2<Double, Integer> createAccumulator() {
            return new Tuple2<>(0.0, 0);
        }

        // 必须实现一个accumulate方法，来数据之前更新状态
        public void accumulate(Tuple2<Double, Integer> accumulater, Double temp) {
            accumulater.f0 += temp;
            accumulater.f1 += 1;
        }
    }
}
